package com.etc.sparkstrea

import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStrea{
  def StreamingWordCount(): Unit ={
    val conf = new SparkConf().setAppName("StreamingWordCount").setMaster("local[*]")
    val sc = new StreamingContext(conf,Seconds(1))
    val line = sc.socketTextStream("192.168.88.5",8888)
    line.flatMap(_.split(" "))
      .map(a => (a,1))
      .reduceByKey(_+_)
      .print()
    sc.start()
    sc.awaitTermination()
  }

  def StreamingHdfs(): Unit ={
    val conf = new SparkConf().setAppName("StreamingHdfs").setMaster("local[*]")
    val sc = new StreamingContext(conf,Seconds(1))
    val line = sc.textFileStream("hdfs://master:9000/input")
    line.flatMap(_.split(","))
      .map(a => (a,1))
      .reduceByKey(_+_)
      .print()
    sc.start()
    sc.awaitTermination()
  }
  def StreamingKafka(): Unit ={
    val conf = new SparkConf().setAppName("StreamingKafka").setMaster("local[*]")
    val sc = new StreamingContext(conf,Seconds(1))
    //topic名字  分区
    val map = Map(("test",1))
    //zookeeper机台IP
    val value = KafkaUtils.createStream(sc,"192.168.88.5:2181,192.168.88.6:2181,192.168.88.7:2181", "test-consumer-group",map)

    value.flatMap(_._2.split("\t"))
      .map(a => (a,1))
      .reduceByKey(_+_)
      .print()
    sc.start()
    sc.awaitTermination()
  }


  def main(args: Array[String]): Unit = {
//    StreamingWordCount()
//    StreamingHdfs()
    StreamingKafka()
  }
}
